<html>
 <head>
  <meta charset="UTF-8">
 </head>
 <body>
  <h1 data-lake-id="Dh8aX" id="Dh8aX"><span data-lake-id="u27259b2d" id="u27259b2d">典型回答</span></h1>
  <p data-lake-id="u1877ec3f" id="u1877ec3f"><br></p>
  <p data-lake-id="u5a160995" id="u5a160995"><span data-lake-id="u63b7c5dc" id="u63b7c5dc">和Kafka只支持同一个Partition内消息的顺序性一样，RocketMQ中也提供了基于队列(分区)的顺序消费。即同一个队列内的消息可以做到有序，但是不同队列内的消息是无序的！</span></p>
  <p data-lake-id="uf534574f" id="uf534574f"><span data-lake-id="ua64e831d" id="ua64e831d">​</span><br></p>
  <p data-lake-id="u5ecc2c20" id="u5ecc2c20"><span data-lake-id="u07a694a1" id="u07a694a1">当我们作为MQ的生产者需要发送顺序消息时，</span><strong><span data-lake-id="u3215a8fb" id="u3215a8fb">需要在send方法中，传入一个MessageQueueSelector。</span></strong></p>
  <p data-lake-id="u1e2d1d48" id="u1e2d1d48"><span data-lake-id="u48f8d69e" id="u48f8d69e">​</span><br></p>
  <p data-lake-id="u85da0363" id="u85da0363"><span data-lake-id="u6ce8cc4d" id="u6ce8cc4d">MessageQueueSelector中需要实现一个select方法，这个方法就是用来定义要把消息发送到哪个MessageQueue的，通常可以使用取模法进行路由：</span></p>
  <p data-lake-id="u8070b1ef" id="u8070b1ef"><span data-lake-id="u86a1e899" id="u86a1e899">​</span><br></p>
  <pre lang="java"><code>
   SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    @Override
  	//mqs：该Topic下所有可选的MessageQueue
    //msg：待发送的消息
    //arg：发送消息时传递的参数
    public MessageQueue select(List&lt;MessageQueue&gt; mqs, Message msg, Object arg) {
     Integer id = (Integer) arg;
     //根据参数，计算出一个要接收消息的MessageQueue的下标
     int index = id % mqs.size();
     //返回这个MessageQueue
     return mqs.get(index);
    }
   }, orderId);
</code></pre>
  <p data-lake-id="u23ed0c8c" id="u23ed0c8c"><br></p>
  <p data-lake-id="u5d37d31f" id="u5d37d31f"><strong><span data-lake-id="u2da45e06" id="u2da45e06">通过以上形式就可以将需要有序的消息发送到同一个队列中</span></strong><span data-lake-id="u9ef13856" id="u9ef13856">。需要注意的时候，这里需要使用同步发送的方式！</span></p>
  <p data-lake-id="u2894e7be" id="u2894e7be"><span data-lake-id="u2e7d3d12" id="u2e7d3d12">​</span><br></p>
  <p data-lake-id="u4c3fa9bf" id="u4c3fa9bf"><span data-lake-id="u4b8224d2" id="u4b8224d2">消息按照顺序发送的消息队列中之后，那么，消费者如何按照发送顺序进行消费呢？</span></p>
  <p data-lake-id="uadc3fa53" id="uadc3fa53"><span data-lake-id="udb0e8dad" id="udb0e8dad">​</span><br></p>
  <p data-lake-id="u08962695" id="u08962695"><span data-lake-id="u670e7f51" id="u670e7f51">RocketMQ的MessageListener回调函数提供了两种消费模式，有序消费模式MessageListenerOrderly和并发消费模式MessageListenerConcurrently。所以，</span><strong><span data-lake-id="u4fc455ae" id="u4fc455ae">想要实现顺序消费，需要使用MessageListenerOrderly模式接收消息</span></strong><span data-lake-id="u45a9a77c" id="u45a9a77c">：</span></p>
  <p data-lake-id="ucefca901" id="ucefca901"><span data-lake-id="uf5ed325b" id="uf5ed325b">​</span><br></p>
  <pre lang="java"><code>
consumer.registerMessageListener(new MessageListenerOrderly() {
        Override
        public ConsumeOrderlyStatus consumeMessage(List&lt;MessageExt&gt; msgs ,ConsumeOrderlyContext context) {
             System.out.printf("Receive order msg:" + new String(msgs.get(0).getBody()));
             return ConsumeOrderlyStatus.SUCCESS ; 
        }
});
</code></pre>
  <p data-lake-id="u9d1d8b05" id="u9d1d8b05"><span data-lake-id="u834e200c" id="u834e200c">​</span><br></p>
  <p data-lake-id="ufc0f66c8" id="ufc0f66c8"><span data-lake-id="ue46b27aa" id="ue46b27aa">当我们用以上方式注册一个消费之后，为了保证同一个队列中的有序消息可以被顺序消费，就要保证RocketMQ的Broker只会把消息发送到同一个消费者上，这时候就需要</span><strong><span data-lake-id="ue7616494" id="ue7616494">加锁</span></strong><span data-lake-id="uedd1c336" id="uedd1c336">了。</span></p>
  <p data-lake-id="u830537b2" id="u830537b2"><span data-lake-id="ufce47a12" id="ufce47a12">​</span><br></p>
  <p data-lake-id="ud21ee308" id="ud21ee308"><span data-lake-id="u5adeb53a" id="u5adeb53a">在实现中，ConsumeMessageOrderlyService 初始化的时候，会启动一个定时任务，</span><strong><span data-lake-id="u97a332ed" id="u97a332ed">会尝试向 Broker 为当前消费者客户端申请分布式锁</span></strong><span data-lake-id="u57d90f87" id="u57d90f87">。如果获取成功，那么后续消息将会只发给这个Consumer。</span></p>
  <p data-lake-id="u2ad4f090" id="u2ad4f090"><span data-lake-id="u5dfdd582" id="u5dfdd582">​</span><br></p>
  <p data-lake-id="u43ffe796" id="u43ffe796"><span data-lake-id="u7dec9999" id="u7dec9999">接下来在消息拉取的过程中，消费者会一次性拉取多条消息的，并且会将拉取到的消息放入 ProcessQueue，同时将消息提交到消费线程池进行执行。</span></p>
  <p data-lake-id="ue4a37329" id="ue4a37329"><span data-lake-id="u912c8cd3" id="u912c8cd3">​</span><br></p>
  <p data-lake-id="ud5cc5041" id="ud5cc5041"><span data-lake-id="uc02fa120" id="uc02fa120">那么拉取之后的消费过程，怎么保证顺序消费呢？这里就需要更多的锁了。</span></p>
  <p data-lake-id="u129fb35d" id="u129fb35d"><span data-lake-id="uaaf5f169" id="uaaf5f169">​</span><br></p>
  <p data-lake-id="u884e8d6f" id="u884e8d6f"><span data-lake-id="uaaaae7cf" id="uaaaae7cf">RocketMQ在消费的过程中，需要</span><strong><span data-lake-id="u5ea44b70" id="u5ea44b70">申请 MessageQueue 锁</span></strong><span data-lake-id="u880d2388" id="u880d2388">，确保在同一时间，一个队列中只有一个线程能处理队列中的消息。</span></p>
  <p data-lake-id="u941a836e" id="u941a836e"><span data-lake-id="u7aab9251" id="u7aab9251">​</span><br></p>
  <p data-lake-id="ua7b7fad3" id="ua7b7fad3"><span data-lake-id="u2fe6e2e0" id="u2fe6e2e0">获取到 MessageQueue 的锁后，就可以从ProcessQueue中依次拉取一批消息处理了，但是这个过程中，为了保证消息不会出现重复消费，还需要</span><strong><span data-lake-id="u1faa8354" id="u1faa8354">对ProcessQueue进行加锁</span></strong><span data-lake-id="ud58e565a" id="ud58e565a">。（这个在扩展知识中展开）</span></p>
  <p data-lake-id="ua0ce79fe" id="ua0ce79fe"><span data-lake-id="u370bcd06" id="u370bcd06">​</span><br></p>
  <p data-lake-id="u3748a234" id="u3748a234"><span data-lake-id="u4c4b07d4" id="u4c4b07d4">然后就可以开始处理业务逻辑了。</span></p>
  <p data-lake-id="uee2c2fed" id="uee2c2fed"><span data-lake-id="u10bb57e9" id="u10bb57e9">​</span><br></p>
  <p data-lake-id="ufe6132c0" id="ufe6132c0"><strong><span data-lake-id="u09cb3013" id="u09cb3013">总结下来就是三次加锁，先锁定Broker上的MessageQueue，确保消息只会投递到唯一的消费者，对本地的MessageQueue加锁，确保只有一个线程能处理这个消息队列。对存储消息的ProcessQueue加锁，确保在重平衡的过程中不会出现消息的重复消费。</span></strong></p>
  <p data-lake-id="ue87c7352" id="ue87c7352"><span data-lake-id="u57cd1f90" id="u57cd1f90">​</span><br></p>
  <p data-lake-id="uacbd3d50" id="uacbd3d50"><span data-lake-id="ucf09d4bd" id="ucf09d4bd">（完整的处理流程大家可以看一下这张图，是极客时间上某个专栏中的内容，虽然专栏中这段文字描述不太容易懂，但是这个图画的还是挺清晰的。）</span></p>
  <p data-lake-id="u71c696b9" id="u71c696b9"><span data-lake-id="u8ed707bb" id="u8ed707bb">​</span><br></p>
  <p data-lake-id="u0665a0f4" id="u0665a0f4"><img src="https://cdn.nlark.com/yuque/0/2023/png/5378072/1679051138277-fa95bc6f-aacb-4897-9ec3-e08995702699.png?x-oss-process=image%2Fwatermark%2Ctype_d3F5LW1pY3JvaGVp%2Csize_55%2Ctext_SmF2YSA4IEd1IFA%3D%2Ccolor_FFFFFF%2Cshadow_50%2Ct_80%2Cg_se%2Cx_10%2Cy_10"></p>
  <p data-lake-id="u28936916" id="u28936916"><span data-lake-id="u0fb62ab2" id="u0fb62ab2">​</span><br></p>
  <p data-lake-id="u25246059" id="u25246059"><span data-lake-id="uf8af1146" id="uf8af1146">​</span><br></p>
  <h1 data-lake-id="nGAm6" id="nGAm6"><span data-lake-id="u7885ae7c" id="u7885ae7c">扩展知识</span></h1>
  <p data-lake-id="udc81f33c" id="udc81f33c"><br></p>
  <h2 data-lake-id="SyFGZ" id="SyFGZ"><span data-lake-id="u64e31ec1" id="u64e31ec1">第三把锁有什么用？</span></h2>
  <p data-lake-id="u279693e4" id="u279693e4"><br></p>
  <p data-lake-id="udc30ee54" id="udc30ee54"><span data-lake-id="u542af05a" id="u542af05a">前面介绍客户端加锁过程中，一共加了三把锁，那么，有没有想过这样一个问题，第三把锁如果不加的话，是不是也没问题？</span></p>
  <p data-lake-id="u3fe55b17" id="u3fe55b17"><span data-lake-id="u315196e2" id="u315196e2">​</span><br></p>
  <p data-lake-id="ue96f9b2d" id="ue96f9b2d"><span data-lake-id="u77c54c1e" id="u77c54c1e">因为我们已经对</span><span data-lake-id="u48acdc96" id="u48acdc96" class="lake-fontsize-12" style="color: rgb(51, 51, 51)">MessageQueue加锁了，为啥还需要对</span><span data-lake-id="u9e24f45d" id="u9e24f45d" style="color: rgb(18, 18, 18)">ProcessQueue再次加锁呢？</span></p>
  <p data-lake-id="u4622ef88" id="u4622ef88"><span data-lake-id="u87b06c07" id="u87b06c07" style="color: rgb(18, 18, 18)">​</span><br></p>
  <p data-lake-id="u04119db9" id="u04119db9"><span data-lake-id="ud2108104" id="ud2108104" style="color: rgb(18, 18, 18)">这里其实主要考虑的是</span><strong><span data-lake-id="uac7fd01a" id="uac7fd01a" style="color: rgb(18, 18, 18)">重平衡的问题</span></strong><span data-lake-id="u3777f50b" id="u3777f50b" style="color: rgb(18, 18, 18)">。</span></p>
  <p data-lake-id="u75dbd228" id="u75dbd228"><span data-lake-id="u5d0cc590" id="u5d0cc590" style="color: rgb(18, 18, 18)">​</span><br></p>
  <p data-lake-id="u3cae109b" id="u3cae109b"><span data-lake-id="u35683f08" id="u35683f08">当我们的消费者集群，新增了一些消费者，发生重平衡的时候，某个队列可能会原来属于客户端A消费的，但是现在要重新分配给客户端B了。</span></p>
  <p data-lake-id="u3bb5cad5" id="u3bb5cad5"><span data-lake-id="uf01602a5" id="uf01602a5">​</span><br></p>
  <p data-lake-id="u61f94804" id="u61f94804"><span data-lake-id="uf6af7d2b" id="uf6af7d2b">这时候客户端A就需要把自己加在Broker上的锁解掉，而在这个解锁的过程中，就需要确保消息不能在消费过程中就被移除了，因为如果客户端A可能</span><span data-lake-id="u1800022f" id="u1800022f">正在处理一部分消息，但是位点信息还没有提交，如果客户端B立马去消费队列中的消息，那存在一部分数据会被重复消费。</span></p>
  <p data-lake-id="u93d7d9d3" id="u93d7d9d3"><span data-lake-id="u12ceec49" id="u12ceec49">​</span><br></p>
  <p data-lake-id="u96bb7bf2" id="u96bb7bf2"><span data-lake-id="u4201cfbb" id="u4201cfbb">那么如何判断消息是否正在消费中呢，</span><strong><span data-lake-id="u285cd734" id="u285cd734">就需要通过这个ProcessQueue上面的锁来判断了，也就是说在解锁的线程也需要尝试对ProcessQueue进行加锁，加锁成功才能进行解锁操作。以避免过程中有消息消费。</span></strong></p>
  <p data-lake-id="u076fc2c4" id="u076fc2c4"><span data-lake-id="ucf40aa6b" id="ucf40aa6b"><br><br></span></p>
  <h2 data-lake-id="tq9Iw" id="tq9Iw"><span data-lake-id="ufbfd9582" id="ufbfd9582">顺序消费存在的问题</span></h2>
  <p data-lake-id="u68ba5cc1" id="u68ba5cc1"><br></p>
  <p data-lake-id="u628ff027" id="u628ff027"><span data-lake-id="u6cca30dc" id="u6cca30dc">通过上面的介绍，我们知道了RocketMQ的顺序消费是通过在消费者上多次加锁实现的，这种方式带来的问题就是会降低吞吐量，并且如果前面的消息阻塞，会导致更多消息阻塞。所以，顺序消息需要慎用。</span></p>
  <p data-lake-id="u33ddb7ed" id="u33ddb7ed"><span data-lake-id="u163997e9" id="u163997e9">​</span><br></p>
  <p data-lake-id="u47c96d31" id="u47c96d31"><span data-lake-id="uc039c49f" id="uc039c49f">​</span><br></p>
  <h2 data-lake-id="Jvzib" id="Jvzib"><span data-lake-id="u7795f5b1" id="u7795f5b1"><br><br></span></h2>
 </body>
</html>